package com.ld.producer.interceptor;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * @author ld
 *
 * 生产者拦截器的生产者客户端
 */
public class ProducerInterceptorKafkaProducer {

    //kafka服务器地址
    private static String servers = "192.168.150.101:9092,192.168.150.102:9092,192.168.150.103:9092";
    //客户端id
    private static String clientId = "product.client.producer.interceptor";
    //主题名称
    private static String topic = "first";


    /**
     * 生产者客户端配置
     * 这里只配置了几个必要的参数，其他参数后面会一一介绍。
     * @return
     */
    private static Properties config(){
        Properties properties = new Properties();
        //kafka服务器地址
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,servers);
        //key的序列化器类路径,序列化器可以直接获取类路径
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //value的序列化器类路径
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
        //客户端id
        properties.put(ProducerConfig.CLIENT_ID_CONFIG,clientId);

        //配置生产者拦截器,还可以配置多个拦截器（以逗号分隔）,形成拦截器链,拦截器链中拦截器的执行顺序是按照配置的顺序执行。
        properties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,MyProducerInterceptor.class.getName());

        //返回配置对象
        return properties;
    }

    public static void main(String[] args) {
        //获取客户端配置
        Properties properties = config();
        //创建KafkaProducer对象,传入客户端配置信息
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
        for(int i = 0;i < 10;i++){
            //创建ProducerRecord对象,传入主题名称，消息
            ProducerRecord<String, String> record = new ProducerRecord<>(topic,"消息" + i);

            //发送ProducerRecord对象
            try{
                //普通发送
                producer.send(record);
            }catch (Exception e){
                e.printStackTrace();
            }
        }
        //关闭KafkaProducer对象
        producer.close();
    }

}
